package com.zmj.tl.database.taos.highvolume;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.zmj.tl.database.taos.config.TdengineConfig;
import com.zmj.tl.database.taos.listener.IotKafkaListener;
import lombok.SneakyThrows;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Resource;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * @author zhangjp
 * @version 1.0
 * @title FastWriteExample
 * @description
 * @create 2024/4/10 10:22
 */

public class FastWriteExample {

    final static Logger logger = LoggerFactory.getLogger(FastWriteExample.class);

    final static int TASK_QUEUE_CAPACITY = 1000000;
    final static List<BlockingQueue<String>> BLOCKING_QUEUES = new ArrayList<>();
    final static List<ReadTask> READ_TASKS = new ArrayList<>();
    final static List<WriteTask> WRITE_TASKS = new ArrayList<>();
    final static DataBaseMonitor DATA_BASE_MONITOR = new DataBaseMonitor();
    @Resource
    private TdengineConfig tdengineConfig;

    public static void stopAll() {
        logger.info("shutting down");
        READ_TASKS.forEach(ReadTask::stop);
        WRITE_TASKS.forEach(WriteTask::stop);
        DATA_BASE_MONITOR.close();
    }

    public static void main(String[] args) throws InterruptedException, SQLException {
        int readTaskCount = args.length > 0 ? Integer.parseInt(args[0]) : 1;
        int writeTaskCount = args.length > 1 ? Integer.parseInt(args[1]) : 3;
        int tableCount = args.length > 2 ? Integer.parseInt(args[2]) : 1000;
        int maxBatchSize = args.length > 3 ? Integer.parseInt(args[3]) : 3000;

        logger.info("readTaskCount={}, writeTaskCount={} tableCount={} maxBatchSize={}",
                readTaskCount, writeTaskCount, tableCount, maxBatchSize);

        DATA_BASE_MONITOR.init();

        // Create task queues, whiting tasks and start writing threads.
        for (int i = 0; i < writeTaskCount; ++i) {
            BlockingQueue<String> queue = new ArrayBlockingQueue<>(TASK_QUEUE_CAPACITY);
            BLOCKING_QUEUES.add(queue);
            WriteTask task = new WriteTask(queue, maxBatchSize);
            Thread t = new Thread(task);
            t.setName("WriteThread-" + i);
            t.start();
        }

//        // create reading tasks and start reading threads
//        int tableCountPerTask = tableCount / readTaskCount;
//        for (int i = 0; i < readTaskCount; ++i) {
//            ReadTask task = new ReadTask(i, BLOCKING_QUEUES, tableCountPerTask);
//            Thread t = new Thread(task);
//            t.setName("ReadThread-" + i);
//            t.start();
//        }

        String value = "[{\"metric\":\"spt_1702001\",\"timestamp\":1714975910513,\"value\":\"144.07\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_cartgearmotor1_vfdcurrentvalue\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"大车_减速电机1变频器电流值\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_CartGearmotor1_VFDCurrentValue\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910513,\"value\":\"772.06\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_cartgearmotor1_vfdvelocityvalue\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"大车_减速电机1变频器转速值\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_CartGearmotor1_VFDVelocityValue\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910523,\"value\":\"104351.20\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_carttocentercontrol_cartcurrentposition\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"大车To中控_大车当前位置\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_CartToCenterControl_CartCurrentPosition\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910523,\"value\":\"-1.05\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_carttocentercontrol_cartcurrentvelocity\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"大车To中控_大车当前速度\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_CartToCenterControl_CartCurrentVelocity\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910555,\"value\":\"10.23\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_trolleygearmotor1_vfdcurrentvalue\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"小车_减速电机1变频器电流值\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_TrolleyGearmotor1_VFDCurrentValue\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910555,\"value\":\"-10.12\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_trolleygearmotor1_vfdvelocityvalue\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"小车_减速电机1变频器转速值\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_TrolleyGearmotor1_VFDVelocityValue\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910555,\"value\":\"18506.20\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_trolleytocentercontrol_trolleycurrentposition\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"小车To中控_小车当前位置\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_TrolleyToCenterControl_TrolleyCurrentPosition\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910555,\"value\":\"0.01\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_trolleytocentercontrol_trolleycurrentvelocity\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"小车To中控_小车当前速度\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_TrolleyToCenterControl_TrolleyCurrentVelocity\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910577,\"value\":\"0.00\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_hoistmotor1_vfdcurrentvalue\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"起升1电机_变频器电流值\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_Hoistmotor1_VFDCurrentValue\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910592,\"value\":\"0.02\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_hoistmotor2_vfdcurrentvalue\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"起升2电机_变频器电流值\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_Hoistmotor2_VFDCurrentValue\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910592,\"value\":\"177.00\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_hoistoverweightdetection_emptyweight1\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"起升超重检测_空载重量1\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_HoistOverweightDetection_EmptyWeight1\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910592,\"value\":\"177.00\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_hoistweightdetection_totalnetweight\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"起升称重检测_总负载的净重\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_HoistWeightDetection_TotalNetWeight\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910603,\"value\":\"4356.20\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_hoisttocentercontrol_hoistcurrentposition\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"起升To中控_起升当前位置\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_HoistToCenterControl_HoistCurrentPosition\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910603,\"value\":\"0.00\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_hoisttocentercontrol_hoistcurrentvelocity\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"起升To中控_起升当前速度\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_HoistToCenterControl_HoistCurrentVelocity\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910624,\"value\":\"914.73\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_magnet_ultrasoundvalue1\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"电磁吊具超声波_超声波1检测距离值\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_Magnet_UltrasoundValue1\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910670,\"value\":\"105841.81\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_lr_ecmenergymeter_positiveactiveenergy\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"电控模块电表_正向有功电能\",\"message_id\":\"1714975911461\",\"attribute_code\":\"lr_ECMEnergyMeter_PositiveActiveEnergy\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910670,\"value\":\"235.92\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_ecmenergymeter_l1voltage\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"电控模块电表_L1相电压\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_ECMEnergyMeter_L1Voltage\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910670,\"value\":\"235.26\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_ecmenergymeter_l2voltage\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"电控模块电表_L2相电压\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_ECMEnergyMeter_L2Voltage\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910670,\"value\":\"235.52\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_ecmenergymeter_l3voltage\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"电控模块电表_L3相电压\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_ECMEnergyMeter_L3Voltage\",\"quality\":true}},{\"metric\":\"spt_1702001\",\"timestamp\":1714975910670,\"value\":\"143800.34\",\"tags\":{\"tenant_id\":\"1000\",\"device_name\":\"地操双桥梁式起重机\",\"receive_time\":\"2024-05-06 14:13:53\",\"tname\":\"sbt_1702001_r_ecmenergymeter_instantaneoustotalpower\",\"device_no\":\"1702001\",\"device_type\":\"0\",\"attribute_name\":\"电控模块电表_瞬时总功率\",\"message_id\":\"1714975911461\",\"attribute_code\":\"r_ECMEnergyMeter_InstantaneousTotalPower\",\"quality\":true}}]";
        List<Map<String, Object>> lists = JSON.parseObject(JSONUtil.toJsonStr(value), new TypeReference<List<Map<String, Object>>>() {
        });
        for (Map<String, Object> list : lists) {
            FastWriteExample fastWriteExample = new FastWriteExample();
            fastWriteExample.fastWrite(list);
        }
//            ReadTask task = new ReadTask(i, BLOCKING_QUEUES, tableCountPerTask,objectObjectHashMap);
//            Thread t = new Thread(task);
//            t.setName("ReadThread-" + i);
//            t.start();

        Runtime.getRuntime().addShutdownHook(new Thread(FastWriteExample::stopAll));

//        long lastCount = 0;
//        while (true) {
//            Thread.sleep(10000);
//            long numberOfTable = DATA_BASE_MONITOR.getTableCount();
//            long count = DATA_BASE_MONITOR.count();
//            logger.info("numberOfTable={} count={} speed={}", numberOfTable, count, (count - lastCount) / 10);
//            lastCount = count;
//        }
    }

    @SneakyThrows
    private void fastWrite(Map<String, Object> map) {
        int readTaskCount = 1;
        int writeTaskCount = 3;
        int tableCount = 1000;
        int maxBatchSize = 3000;

        logger.info("readTaskCount={}, writeTaskCount={} tableCount={} maxBatchSize={}",
                readTaskCount, writeTaskCount, tableCount, maxBatchSize);
        DATA_BASE_MONITOR.init();


        // Create task queues, whiting tasks and start writing threads.
        for (int i = 0; i < writeTaskCount; ++i) {
            BlockingQueue<String> queue = new ArrayBlockingQueue<>(TASK_QUEUE_CAPACITY);
            BLOCKING_QUEUES.add(queue);
            WriteTask task = new WriteTask(queue, maxBatchSize);
            Thread t = new Thread(task);
            t.setName("WriteThread-" + i);
            t.start();
        }
        // create reading tasks and start reading threads
        int tableCountPerTask = tableCount / readTaskCount;
        for (int i = 0; i < readTaskCount; ++i) {
            ReadTask task = new ReadTask(i, BLOCKING_QUEUES, tableCountPerTask, map);
            Thread t = new Thread(task);
            t.setName("ReadThread-" + i);
            t.start();
        }


        Runtime.getRuntime().addShutdownHook(new Thread(IotKafkaListener::stopAll));

//        long lastCount = 0;
//        while (true) {
//            Thread.sleep(10000);
//            long numberOfTable = DATA_BASE_MONITOR.getTableCount();
//            long count = DATA_BASE_MONITOR.count();
//            logger.info("numberOfTable={} count={} speed={}", numberOfTable, count, (count - lastCount) / 10);
//            lastCount = count;
//        }

    }
}
